﻿#include "service_finder_zookeeper.hh"
#include "../zookeeper/zookeeper_client.hh"
#include <random>
#include <vector>

namespace kratos {
namespace service {

ServiceFinderZookeeper::ServiceFinderZookeeper() {
  range_.seed(std::random_device()());
}

ServiceFinderZookeeper::~ServiceFinderZookeeper() {}

auto ServiceFinderZookeeper::start(const std::string &servers, int timeout)
    -> bool {
  if (zoo_ptr_) {
    if (zoo_ptr_->is_connected()) {
      // 已经连接
      return true;
    }
  }
  if (servers.empty()) {
    return false;
  }
  if (timeout <= 0) {
    // 最小超时为1000毫秒
    timeout = 1000;
  }
  zoo_ptr_ = make_unique_pool_ptr<ZookeeperClient>();
  return zoo_ptr_->connect(servers, timeout);
}

auto ServiceFinderZookeeper::stop() -> void {
  if (!zoo_ptr_) {
    return;
  }
  // 断开连接
  zoo_ptr_->disconnect();
  // 释放, 其他调用函数都将失效
  zoo_ptr_.reset();
}

void ServiceFinderZookeeper::update(std::time_t tick) {
  if (!zoo_ptr_) {
    return;
  }
  if (!tick) {
    last_tick_ = tick;
  }
  if (tick - last_tick_ >= CHECK_INTERVAL) {
    if (!zoo_ptr_->is_connected()) {
      zoo_ptr_->reconnect();
    }
    last_tick_ = tick;
  }
  if (zoo_ptr_->is_changed()) {
    std::vector<std::string> changed_keys;
    zoo_ptr_->get_changed_path(changed_keys);
    for (const auto& key : changed_keys) {
      auto it = listener_map_.find(key);
      if (it == listener_map_.end()) {
        continue;
      } else {
        std::vector<std::string> hosts;
        if (zoo_ptr_->get_children(key, hosts)) {
          it->second(key, hosts);
        }
      }
    }
  }
}

auto ServiceFinderZookeeper::find_service(const std::string &name,
                                          std::list<std::string> &hosts)
    -> bool {
  hosts.clear();
  if (!zoo_ptr_) {
    return false;
  }
  if (name.empty()) {
    return false;
  }
  std::string real_name;
  if (name[0] != '/') {
    real_name = "/" + name;
  } else {
    real_name = name;
  }
  return zoo_ptr_->get_children(real_name, hosts);
}

auto ServiceFinderZookeeper::find_service(const std::string &name,
                                          std::string &host) -> bool {
  // 负载均衡, 均匀分布
  if (!zoo_ptr_) {
    return false;
  }
  if (name.empty()) {
    return false;
  }
  std::string real_name;
  if (name[0] != '/') {
    real_name = "/" + name;
  } else {
    real_name = name;
  }
  std::vector<std::string> hosts;
  if (!zoo_ptr_->get_children(real_name, hosts)) {
    return false;
  }
  if (hosts.empty()) {
    return false;
  }
  std::uniform_int_distribution<std::mt19937::result_type> dice(
      0, static_cast<int>(hosts.size()) - 1);
  host = hosts[dice(range_)];
  return true;
}

auto ServiceFinderZookeeper::dump(std::ostream &ostream) -> void {
  if (!zoo_ptr_) {
    return;
  }
  zoo_ptr_->dump(ostream);
}

auto ServiceFinderZookeeper::remove_cache(const std::string &name,
                                          const std::string &host) -> void {
  if (!zoo_ptr_) {
    return;
  }
  std::string real_name;
  if (name[0] != '/') {
    real_name = "/" + name;
  } else {
    real_name = name;
  }
  zoo_ptr_->remove_cache(real_name, host);
}

auto ServiceFinderZookeeper::get_service_host_count(const std::string &name)
    -> std::size_t {
  if (!zoo_ptr_) {
    return 0;
  }
  return zoo_ptr_->get_children_count(name);
}

auto ServiceFinderZookeeper::add_listener(const std::string& listener_name,
    ServiceChange service_change) -> bool {
  listener_map_[listener_name] = service_change;
  return true;
}

} // namespace service
} // namespace kratos
